/*
 *  Copyright (c) 2017 Red Hat, Inc. and/or its affiliates.
 *  Copyright (c) 2017 INSA Lyon, CITI Laboratory.
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.vertx.guides.wiki;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.Message;
import io.vertx.core.file.FileSystem;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.asyncsql.MySQLClient;
import io.vertx.ext.sql.ResultSet;
import io.vertx.ext.sql.SQLClient;
import io.vertx.ext.sql.SQLConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.UUID;
import java.util.stream.Collectors;


/**
 * 扩展性非常好，通过配置文件来进行数据库操作，随时更换jdbc驱动，不同的sql dialect也可重用该类
 */
public class WikiDatabaseVerticle extends AbstractVerticle {

  public static final String CONFIG_MYSQL_HOST = "127.0.0.1";
  public static final Integer CONFIG_MYSQL_PORT = 3306;
  public static final Integer CONFIG_MYSQL_MAXPOOLSIZE = 100;
  public static final String CONFIG_MYSQL_USERNAME = "root";
  public static final String CONFIG_MYSQL_PASSWORD = "root";
  public static final String CONFIG_MYSQL_DATABASE = "vertx_test";
  public static final String CONFIG_MYSQL_CHARSET = "UTF-8";
  public static final Integer CONFIG_MYSQL_QUERYTIMEOUT = 10000;

  //  public static final String CONFIG_WIKIDB_JDBC_URL = "wikidb.jdbc.url";
//  public static final String CONFIG_WIKIDB_JDBC_DRIVER_CLASS = "wikidb.jdbc.driver_class";
//  public static final String CONFIG_WIKIDB_JDBC_MAX_POOL_SIZE = "wikidb.jdbc.max_pool_size";
  public static final String CONFIG_WIKIDB_SQL_QUERIES_RESOURCE_FILE = "wikidb.sqlqueries.resource.file";

  public static final String CONFIG_WIKIDB_QUEUE = "wikidb.queue";

  private static final Logger LOGGER = LoggerFactory.getLogger(WikiDatabaseVerticle.class);

  // (...)
  // end::preamble[]

  // tag::loadSqlQueries[]
  private enum SqlQuery {
    CREATE_PAGES_TABLE,
    ALL_PAGES,
    GET_PAGE,
    CREATE_PAGE,
    SAVE_PAGE,
    DELETE_PAGE,
    CREATE_USER,
    CREATE_USERREG_TABLE,
    CREATE_USERREG,
    GET_USERREG,
    DELETE_USERREG,
    CREATE_USERLOGIN_TABLE,
    CREATE_USERLOGIN,
    GET_USERLOGIN,
    DELETE_USERLOGIN,
    CREATE_USERINFO_TABLE,
    CREATE_USERINFO,
    GET_USERINFO,
    DELETE_USERINFO
  }

  private final HashMap<SqlQuery, String> sqlQueries = new HashMap<>();

  /**
   * 加载sql配置文件，存在map里
   *
   * @throws IOException
   */
  private void loadSqlQueries() throws IOException {

    String queriesFile = config().getString(CONFIG_WIKIDB_SQL_QUERIES_RESOURCE_FILE);
    InputStream queriesInputStream;
    if (queriesFile != null) {
      queriesInputStream = new FileInputStream(queriesFile);
    } else {
      queriesInputStream = getClass().getResourceAsStream("/db-queries.properties");
    }

    Properties queriesProps = new Properties();
    queriesProps.load(queriesInputStream);
    queriesInputStream.close();

    sqlQueries.put(SqlQuery.CREATE_PAGES_TABLE, queriesProps.getProperty("create-pages-table"));

    sqlQueries.put(SqlQuery.ALL_PAGES, queriesProps.getProperty("all-pages"));
    sqlQueries.put(SqlQuery.GET_PAGE, queriesProps.getProperty("get-page"));
    sqlQueries.put(SqlQuery.CREATE_PAGE, queriesProps.getProperty("create-page"));
    sqlQueries.put(SqlQuery.SAVE_PAGE, queriesProps.getProperty("save-page"));
    sqlQueries.put(SqlQuery.DELETE_PAGE, queriesProps.getProperty("delete-page"));

    sqlQueries.put(SqlQuery.CREATE_USERREG_TABLE, queriesProps.getProperty("create-userreg-table"));
    sqlQueries.put(SqlQuery.CREATE_USERREG, queriesProps.getProperty("create-userreg"));
    sqlQueries.put(SqlQuery.GET_USERREG, queriesProps.getProperty("get-userreg"));
    sqlQueries.put(SqlQuery.DELETE_USERREG, queriesProps.getProperty("delete-userreg"));
    sqlQueries.put(SqlQuery.CREATE_USERLOGIN_TABLE, queriesProps.getProperty("create-userlogin-table"));
    sqlQueries.put(SqlQuery.CREATE_USERLOGIN, queriesProps.getProperty("create-userlogin"));
    sqlQueries.put(SqlQuery.GET_USERLOGIN, queriesProps.getProperty("get-userlogin"));
    sqlQueries.put(SqlQuery.DELETE_USERLOGIN, queriesProps.getProperty("delete-userlogin"));

    sqlQueries.put(SqlQuery.CREATE_USERINFO_TABLE, queriesProps.getProperty("create-userinfo-table"));
    sqlQueries.put(SqlQuery.CREATE_USERINFO, queriesProps.getProperty("create-userinfo"));
    sqlQueries.put(SqlQuery.GET_USERINFO, queriesProps.getProperty("get-userinfo"));
    sqlQueries.put(SqlQuery.DELETE_USERINFO, queriesProps.getProperty("delete-userinfo"));
  }
  // end::loadSqlQueries[]

  // tag::start[]
  private SQLClient dbClient;

  @Override
  public void start(Future<Void> startFuture) throws Exception {

    /*
    *  加载sql配置，这里是同步的，但是数据量小和异步编程没什么差异
     * Note: this uses blocking APIs, but data is small...
     */
    loadSqlQueries();  // <1>
    //配置datasource client
    JsonObject mySQLClientConfig = new JsonObject()
      .put("url", CONFIG_MYSQL_HOST).put("port", CONFIG_MYSQL_PORT)
      .put("maxPoolSize", CONFIG_MYSQL_MAXPOOLSIZE)
      .put("username", CONFIG_MYSQL_USERNAME)
      .put("password", CONFIG_MYSQL_PASSWORD)
      .put("database", CONFIG_MYSQL_DATABASE)
      .put("charset", CONFIG_MYSQL_CHARSET)
      .put("queryTimeout", CONFIG_MYSQL_QUERYTIMEOUT);
    dbClient = MySQLClient.createShared(vertx, mySQLClientConfig,"MySQLPool1");


    executeDDL(sqlQueries.get(SqlQuery.CREATE_PAGES_TABLE))
      .compose(v -> executeDDL(sqlQueries.get(SqlQuery.CREATE_USERREG_TABLE)))
      .compose(v -> executeDDL(sqlQueries.get(SqlQuery.CREATE_USERLOGIN_TABLE)))
      .compose(v -> executeDDL(sqlQueries.get(SqlQuery.CREATE_USERINFO_TABLE))).setHandler(startFuture.completer());

  }

  /**
   * ddl执行方法
   *
   * @param sql
   */
  private Future<Void> executeDDL(String sql) {
    Future<Void> future = Future.future();
    dbClient.getConnection(ar -> {
      if (ar.failed()) {
        LOGGER.error("Could not open a database connection", ar.cause());
        future.fail(ar.cause());
      } else {
        SQLConnection connection = ar.result();
        connection.execute(sql, create -> {   // <2>
          connection.close();
          if (create.failed()) {
            LOGGER.error("Database preparation error", create.cause());
            future.fail(create.cause());
          } else {
            //注册一个handler到event bus，用于处理消费消息的逻辑
            vertx.eventBus().consumer(config().getString(CONFIG_WIKIDB_QUEUE, "wikidb.queue"), this::onMessage);
            future.complete();
          }
        });
      }
      LOGGER.debug("step...");
    });
    return future;
  }

  // tag::onMessage[]
  public enum ErrorCodes {
    NO_ACTION_SPECIFIED,
    BAD_ACTION,
    DB_ERROR
  }

  /**
   * event bus注册handler后，由该方法处理消息
   *
   * @param message
   */

  public void onMessage(Message<JsonObject> message) {

    if (!message.headers().contains("action")) {
      LOGGER.error("No action header specified for message with headers {} and body {}",
        message.headers(), message.body().encodePrettily());
      message.fail(ErrorCodes.NO_ACTION_SPECIFIED.ordinal(), "No action header specified");
      return;
    }
    String action = message.headers().get("action");

    switch (action) {
      case "all-pages":
        fetchAllPages(message);
        break;
      case "get-page":
        fetchPage(message);
        break;
      case "create-page":
        createPage(message);
        break;
      case "save-page":
        savePage(message);
        break;
      case "delete-page":
        deletePage(message);
        break;
      case "create-userreg":
        createUserReg(message);
      case "get-userreg":
        fetchUserReg(message);
      case "delete-userreg":
        deleteUserReg(message);
      case "create-userlogin":
        createUserLogin(message);
      case "get-userlogin":
        fetchUserLogin(message);
      case "delete-userlogin":
        deleteUserLogin(message);
      case "create-userinfo":
        createUserInfo(message);
      case "delete-userinfo":
        deleteUserInfo(message);
        break;
      default:
        message.fail(ErrorCodes.BAD_ACTION.ordinal(), "Bad action: " + action);
    }
  }

  private void deleteUserInfo(Message<JsonObject> message) {
    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray().add(message.body().getString("id"));
        connection.updateWithParams(sqlQueries.get(SqlQuery.DELETE_USERINFO), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });

  }

  private void deleteUserLogin(Message<JsonObject> message) {
    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray().add(message.body().getString("id"));
        connection.updateWithParams(sqlQueries.get(SqlQuery.DELETE_USERLOGIN), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });

  }

  private void deleteUserReg(Message<JsonObject> message) {
    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray().add(message.body().getString("guid"));
        connection.updateWithParams(sqlQueries.get(SqlQuery.DELETE_USERREG), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  private void createUserReg(Message<JsonObject> message) {
    JsonObject request = message.body();
    LOGGER.debug("createUserReg ::: {}", message);
    LOGGER.debug("dbClient:::{}", dbClient);
    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray()
          .add(request.getString("guid"))
          .add(request.getString("username"))
          .add(request.getString("password"))
          .add(request.getInteger("sex") == null ? 0 : request.getInteger("sex"));
        LOGGER.debug("createUserReg {} ",data);
        LOGGER.debug("createUserReg  connection {} ",connection);
        connection.updateWithParams(sqlQueries.get(SqlQuery.CREATE_USERREG), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });

  }

  private void fetchUserReg(Message<JsonObject> message) {
    String guid = message.body().getString("guid");

    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        connection.queryWithParams(sqlQueries.get(SqlQuery.GET_USERREG), new JsonArray().add(guid), fetch -> {
          connection.close();
          if (fetch.succeeded()) {
            JsonObject response = new JsonObject();
            ResultSet resultSet = fetch.result();
            if (resultSet.getNumRows() == 0) {
              response.put("found", false);
            } else {
              response.put("found", true);
              JsonArray row = resultSet.getResults().get(0);
              response.put("guid", row.getString(0));
              response.put("username", row.getString(1));
              response.put("password", row.getString(2));
              response.put("sex", row.getInteger(3));
            }
            message.reply(response);
          } else {
            reportQueryError(message, fetch.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }


  private void createUserLogin(Message<JsonObject> message) {

    JsonObject request = message.body();
    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray()
          .add(request.getString("id"))
          .add(request.getString("guid"))
          .add(request.getString("loginname"))
          .add(request.getString("password"))
          .add(request.getString("nickname")==null?"":request.getString("nickname"))
          .add(request.getInteger("sex") == null ? 0 : request.getInteger("sex"));
        LOGGER.debug("createUserLogin {} ",data);
        LOGGER.debug("createUserLogin  connection {} ",connection);
        connection.updateWithParams(sqlQueries.get(SqlQuery.CREATE_USERLOGIN), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });

  }

  /**
   * 根据loginname查询用户登录信息
   *
   * @param message
   */
  private void fetchUserLogin(Message<JsonObject> message) {
    String loginname = message.body().getString("loginname");

    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        connection.queryWithParams(sqlQueries.get(SqlQuery.GET_USERLOGIN), new JsonArray().add(loginname), fetch -> {
          connection.close();
          if (fetch.succeeded()) {
            JsonObject response = new JsonObject();
            ResultSet resultSet = fetch.result();
            if (resultSet.getNumRows() == 0) {
              response.put("found", false);
            } else {
              response.put("found", true);
              JsonArray row = resultSet.getResults().get(0);
              response.put("id", row.getString(0));
              response.put("guid", row.getString(1));
              response.put("loginname", row.getString(2));
              response.put("password", row.getString(3));
              response.put("nickname", row.getString(4));
              response.put("sex", row.getInteger(5));
            }
            message.reply(response);
          } else {
            reportQueryError(message, fetch.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  private void createUserInfo(Message<JsonObject> message) {
    JsonObject request = message.body();
    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray()
          .add(request.getString("id"))
          .add(request.getString("guid"))
          .add(request.getString("nickname")==null?"":request.getString("nickname"));
        LOGGER.debug("createUserInfo {} ",data);
        LOGGER.debug("createUserInfo  connection {} ",connection);
        connection.updateWithParams(sqlQueries.get(SqlQuery.CREATE_USERINFO), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  /**
   * 根据guid查询用户信息
   *
   * @param message
   */
  private void fetchUserInfo(Message<JsonObject> message) {
    String guid = message.body().getString("guid");

    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        connection.queryWithParams(sqlQueries.get(SqlQuery.GET_USERINFO), new JsonArray().add(guid), fetch -> {
          connection.close();
          if (fetch.succeeded()) {
            JsonObject response = new JsonObject();
            ResultSet resultSet = fetch.result();
            if (resultSet.getNumRows() == 0) {
              response.put("found", false);
            } else {
              response.put("found", true);
              JsonArray row = resultSet.getResults().get(0);
              response.put("id", row.getInteger(0));
              response.put("guid", row.getString(1));
              response.put("nickname", row.getString(2));
            }
            message.reply(response);
          } else {
            reportQueryError(message, fetch.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });

  }

  // end::onMessage[]

  // tag::rest[]
  private void fetchAllPages(Message<JsonObject> message) {

    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        connection.query(sqlQueries.get(SqlQuery.ALL_PAGES), res -> {
          connection.close();
          if (res.succeeded()) {
            List<String> pages = res.result()
              .getResults()
              .stream()
              .map(json -> json.getString(0))
              .sorted()
              .collect(Collectors.toList());
            message.reply(new JsonObject().put("pages", new JsonArray(pages)));
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }


  private void fetchPage(Message<JsonObject> message) {

    String requestedPage = message.body().getString("page");

    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        connection.queryWithParams(sqlQueries.get(SqlQuery.GET_PAGE), new JsonArray().add(requestedPage), fetch -> {
          connection.close();
          if (fetch.succeeded()) {
            JsonObject response = new JsonObject();
            ResultSet resultSet = fetch.result();
            if (resultSet.getNumRows() == 0) {
              response.put("found", false);
            } else {
              response.put("found", true);
              JsonArray row = resultSet.getResults().get(0);
              response.put("id", row.getInteger(0));
              response.put("rawContent", row.getString(1));
            }
            message.reply(response);
          } else {
            reportQueryError(message, fetch.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  private void createUser(Message<JsonObject> message) {
    JsonObject request = message.body();

    dbClient.getConnection(car -> {
      LOGGER.info("ddddd ");
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray()
          .add(UUID.randomUUID().toString().replace("-", ""))
          .add(request.getString("username"))
          .add(request.getString("password"));
        LOGGER.debug(data.toString());
        System.out.print(data);
        LOGGER.debug("create user:::" + data.toString());
        connection.updateWithParams(sqlQueries.get(SqlQuery.CREATE_USER), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  private void createPage(Message<JsonObject> message) {
    JsonObject request = message.body();

    dbClient.getConnection(car -> {

      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray()
          .add(request.getString("title"))
          .add(request.getString("markdown"));

        connection.updateWithParams(sqlQueries.get(SqlQuery.CREATE_PAGE), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  private void savePage(Message<JsonObject> message) {
    JsonObject request = message.body();

    dbClient.getConnection(car -> {

      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray()
          .add(request.getString("markdown"))
          .add(request.getString("id"));

        connection.updateWithParams(sqlQueries.get(SqlQuery.SAVE_PAGE), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  private void deletePage(Message<JsonObject> message) {
    dbClient.getConnection(car -> {
      if (car.succeeded()) {
        SQLConnection connection = car.result();
        JsonArray data = new JsonArray().add(message.body().getString("id"));
        connection.updateWithParams(sqlQueries.get(SqlQuery.DELETE_PAGE), data, res -> {
          connection.close();
          if (res.succeeded()) {
            message.reply("ok");
          } else {
            reportQueryError(message, res.cause());
          }
        });
      } else {
        reportQueryError(message, car.cause());
      }
    });
  }

  private void reportQueryError(Message<JsonObject> message, Throwable cause) {
    LOGGER.error("Database query error", cause);
    message.fail(ErrorCodes.DB_ERROR.ordinal(), cause.getMessage());
  }
  // end::rest[]
}
